package com.etlmaster.executor.processors.Table;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.DpExecutorUtils;
import com.etlmaster.executor.utils.JsonHelper;


/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class CreateMPPTableProcessor extends CreateTableProcessor {
	private Map<String, Object> tableMeta = new HashMap<String, Object>();

	@SuppressWarnings("unchecked")
	public void buildDDl() {
		String ddlTemplate = DataCollectionUtils.getProperties(meta.getDsType());
		if (cfgObj.get("extend_cfg") != null && cfgObj.get("extend_cfg").toString().length() > 0) {
			parseExtendCfg();
		}
		Map<String, Object> map = getColumnsInfos(meta.getDsType(), (List<Map<String, Object>>) cfgObj.get("columns"));
		tableMeta.putAll(cfgObj);
		tableMeta.put("columns", map.get("columns"));
		tableMeta.put("partitions", map.get("partitions"));
		tableMeta.put("partyList", map.get("partyList"));
		tableMeta.put("orderby", map.get("orderby"));
		tableMeta.put("key_seq", map.get("key_seq"));
		tableMeta.put("segment_seq", map.get("segment_seq"));
		tableMeta.put("colComments", map.get("comments"));
		tableMeta.put("tableName", meta.getTableName());
		tableMeta.put("options", ((CreateMPPTableMeta) meta).getOptions());
		tableMeta.put("errLog", ((CreateMPPTableMeta) meta).getErrLog());
		tableMeta.put("log", ((CreateMPPTableMeta) meta).getLog());
		tableMeta.put("reject", ((CreateMPPTableMeta) meta).getReject());
		tableMeta.put("serverName", ((CreateMPPTableMeta) meta).getServerName());
		if (meta.getExtraParam() != null)
			tableMeta.putAll(meta.getExtraParam());
		tableMeta.putAll(meta.variables);
		ddl = DpExecutorUtils.variableSubstitution(ddlTemplate, tableMeta);
	}

	public Map<String, String> sortList(String dsType, String colname, String key, List<String> partySeqList,
			List<Map<String, Object>> partyList) {
		Map<String, String> result = new HashMap<String, String>();
		Object[] partySeqArr = partySeqList.toArray();
		Arrays.sort(partySeqArr);
		String partitions = "";

		for (Object obj : partySeqArr) {
			for (Map<String, Object> row : partyList) {
				if (obj.toString().equals(row.get(colname).toString())) {

					partitions += row.get("colname");
					if (dsType.equalsIgnoreCase("hive")) {
						partitions += " " + row.get("datatype");
						if (row.get("length") != null && row.get("length").toString().length() > 0) {
							partitions += "(" + row.get("length");
							if (row.get("precision_val") != null && row.get("precision_val").toString().length() > 0) {
								partitions += "," + row.get("precision_val");
							}
							partitions += ")";
						}
					}
					partitions += ",";
				}
			}
		}
		if (partitions.length() > 0) {
			result.put(key, partitions.substring(0, partitions.length() - 1));
		} else {
			result.put(key, partitions);
		}
		return result;
	}

	private Map<String, Object> getColumnsInfos(String dsType, List<Map<String, Object>> columns) {
		Map<String, Object> result = new HashMap<String, Object>();
		List<Map<String, Object>> newColumns = new ArrayList<Map<String, Object>>();
		List<Map<String, Object>> comments = new ArrayList<Map<String, Object>>();
		List<Map<String, Object>> partyList = new ArrayList<Map<String, Object>>();
		List<Map<String, Object>> orderList = new ArrayList<Map<String, Object>>();
		List<Map<String, Object>> segmentList = new ArrayList<Map<String, Object>>();

		List<Map<String, Object>> keyList = new ArrayList<Map<String, Object>>();
		List<String> partySeqList = new ArrayList<String>();
		List<String> orderSeqList = new ArrayList<String>();
		List<String> segmentSeqList = new ArrayList<String>();

		for (Map<String, Object> column : columns) {
			if (dsType.equalsIgnoreCase("hive")) {
				if (column.get("party_seq") != null && !column.get("party_seq").equals("")) {
					partySeqList.add(column.get("party_seq").toString());
					partyList.add(column);
				} else {
					newColumns.add(column);
				}
			} else {
				newColumns.add(column);
				if (column.get("order_seq") != null && !column.get("order_seq").equals("")) {
					orderSeqList.add(column.get("order_seq").toString());
					orderList.add(column);
				}

				if (column.get("key_seq") != null && !column.get("key_seq").equals("0")) {
					keyList.add(column);
				}
				if (column.get("party_seq") != null && !column.get("party_seq").equals("")) {
					partySeqList.add(column.get("party_seq").toString());
					partyList.add(column);
				}
				if (column.get("segment_seq") != null && !column.get("segment_seq").equals("")) {
					segmentSeqList.add(column.get("segment_seq").toString());
					segmentList.add(column);
				}
				if (column.get("colcomment") != null && !column.get("colcomment").equals("")
						&& !column.get("colcomment").equals("无")) {
					comments.add(column);
				}
			}
		}
		result.put("columns", newColumns);
		result.put("comments", comments);

		result.put("key_seq", keyList);
		result.putAll(sortList(dsType, "party_seq", "partitions", partySeqList, partyList));
		result.putAll(sortList(dsType, "order_seq", "orderby", orderSeqList, orderList));
		result.putAll(sortList(dsType, "segment_seq", "segment_seq", segmentSeqList, segmentList));
		result.put("partyList", partyList);
		return result;
	}

	@SuppressWarnings("unchecked")
	private void parseExtendCfg() {
		Map<String, Object> extendCfgObj = (Map<String, Object>) JsonHelper.getInstance()
				.read(cfgObj.get("extend_cfg").toString());
		tableMeta.putAll(extendCfgObj);
	}
}
